package com.burning.demo.flink.TableAPIAndSQL;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author cai
 * @date 2021/9/3 13:44
 */
public class FlinkSqlSyncMysqlTest {

    public static void main(String[] args) throws Exception {
        testSync(args);
    }

    public static void testSync(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        String sourceSql = "CREATE TABLE table_source(\n" +
                "    id varchar,\n" +
                "    platform_code varchar,\n" +
                "    platform_name varchar,\n" +
                "    visit_count int,\n" +
                "    subscribe_count int,\n" +
                "    favorite_count int,\n" +
                "    share_count int,\n" +
                "    statistic_date timestamp,\n" +
                "    extention varchar,\n" +
                "    mask int,\n" +
                "    ctime timestamp,\n" +
                "    mtime timestamp\n" +
                ") WITH(\n" +
                "    'connector' = 'jdbc',\n" +
                "    'url' = 'jdbc:mysql://47.117.34.236:3306/bidsun_ecs?characterEncoding=UTF8',\n" +
                "    'username' = 'services',\n" +
                "    'password' = '#MYsBvVB',\n" +
                "    'table-name' = 'ads_gzh_data_application_daily'\n" +
//                "    'lookup.cache.max-rows' = '10000',\n" +
//                "    'lookup.cache.ttl' = '1'\n" +
                ")";

        tEnv.executeSql(sourceSql);

        String destSql = "CREATE TABLE table_dest(\n" +
                "    id varchar,\n" +
                "    platform_code varchar,\n" +
                "    platform_name varchar,\n" +
                "    visit_count int,\n" +
                "    subscribe_count int,\n" +
                "    favorite_count int,\n" +
                "    share_count int,\n" +
                "    statistic_date timestamp,\n" +
                "    extention varchar,\n" +
                "    mask int,\n" +
                "    ctime timestamp,\n" +
                "    mtime timestamp\n" +
                ") WITH(\n" +
                "    'connector' = 'jdbc',\n" +
                "    'url' = 'jdbc:mysql://47.117.34.236:3306/bidsun_ecs?characterEncoding=UTF8',\n" +
                "    'username' = 'services',\n" +
                "    'password' = '#MYsBvVB',\n" +
                "    'table-name' = 'ads_gzh_data_application_daily_test'\n" +
                ")";
        tEnv.executeSql(destSql);

        String syncSql = "insert into\n" +
                "    table_dest\n" +
                "select\n" +
                "    id,\n" +
                "    platform_code,\n" +
                "    platform_name,\n" +
                "    visit_count,\n" +
                "    subscribe_count,\n" +
                "    favorite_count,\n" +
                "    share_count,\n" +
                "    statistic_date,\n" +
                "    extention,\n" +
                "    mask,\n" +
                "    ctime,\n" +
                "    mtime\n" +
                "from\n" +
                "    table_source\n" +
                "where\n" +
                "    platform_code = '20210406112059828952424772272128'";
        tEnv.executeSql(syncSql);


    }

}
